package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * com.atguigu.realtime.app.dwd.db.Dwd_09_DwdTradeOrderRefund
 * 交易订单退款
 */
public class Dwd_09_DwdTradeOrderRefund extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_09_DwdTradeOrderRefund().init(
                3009,
                2,
                "Dwd_09_DwdTradeOrderRefund"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_09_DwdTradeOrderRefund");
        // 2. 读取字典表
        readBaseDic(tEnv);

        // 3. 过滤退单表
        Table orderRefundInfo = tEnv.sqlQuery("select \n" +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['refund_type'] refund_type, " +
                "data['refund_num'] refund_num, " +
                "data['refund_amount'] refund_amount, " +
                "data['refund_reason_type'] refund_reason_type, " +
                "data['refund_reason_txt'] refund_reason_txt, " +
                "data['create_time'] create_time, " +
                "pt, " +// " `pt` as proctime() " +  // lookup join的时候要用
                "ts " +
                "from ods_db \n" +
                "where `database`='gmall2022' \n" +
                "and `table`='order_refund_info' \n" +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // 4.过滤订单表-新增一条退单信息，订单表会update
        Table orderInfo = tEnv.sqlQuery("select \n" +
                "data['id'] id, \n" +
                "data['province_id'] province_id, \n" +
                "`old` " +
                "from ods_db \n" +
                "where `database`='gmall2022' \n " +
                "and `table`='order_info' \n" +
                "and `type`='update' \n" +
                "and `old`['order_status'] is not null \n" +
                "and `data`['order_status'] = '1005' ");

        tEnv.createTemporaryView("order_info", orderInfo);

        // 5. 退单 订单 字典 3张表进行join
        Table result = tEnv.sqlQuery("select \n" +
                "ri.id, \n" +//退单id
                "ri.user_id, \n" +
                "ri.order_id, \n" +
                "ri.sku_id, \n" +
                "oi.province_id, \n" +
                "date_format(ri.create_time,'yyyy-MM-dd') date_id, \n" +
                "ri.create_time, \n" +
                "ri.refund_type, \n" +
                "dic1.dic_name, \n" +
                "ri.refund_reason_type, \n" +
                "dic2.dic_name, \n" +
                "ri.refund_reason_txt, \n" +
                "ri.refund_num, \n" +
                "ri.refund_amount, \n" +
                "ri.ts, \n" +
                "current_row_timestamp() row_op_ts \n" +
                "from order_refund_info ri \n" +
                "join order_info oi on ri.order_id=oi.id \n" +
                "join base_dic for system_time as of ri.pt as dic1 on ri.refund_type=dic1.dic_code \n" +
                "join base_dic for system_time as of ri.pt as dic2 on ri.refund_reason_type=dic2.dic_code \n " );

        // 6. 写出到到kafka中
        tEnv.executeSql("create table dwd_trade_order_refund(" +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "sku_id string, " +
                "province_id string, " +
                "date_id string, " +
                "create_time string, " +
                "refund_type_code string, " +
                "refund_type_name string, " +
                "refund_reason_type_code string, " +
                "refund_reason_type_name string, " +
                "refund_reason_txt string, " +
                "refund_num string, " +
                "refund_amount string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));

        result.executeInsert("dwd_trade_order_refund");
    }
}
